package com.baidu.mpks.statistic.elasticSearch;

import com.baidu.mpks.statistic.exception.EsRespException;
import com.baidu.mpks.statistic.exception.StatisticCountException;
import com.baidu.mpks.statistic.meta.Action;
import com.baidu.mpks.statistic.meta.DateCount;
import com.baidu.mpks.utils.JacksonUtils;
import com.baidu.mpks.utils.Utils;
import com.fasterxml.jackson.databind.JsonNode;
import org.elasticsearch.action.search.MultiSearchResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.ShardSearchFailure;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.bucket.range.Range;
import org.elasticsearch.search.aggregations.metrics.Cardinality;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author wangjianhui01
 */
@Service
public class EsService {
    @Resource
    private EsBaseQuery esBaseQuery;

    public static final String SERVICE_KEY = "uri.keyword";
    public static final String FIELD_QUERY = "query.keyword";
    public static final String FIELD_USERID = "userId.keyword";
    public static final String FIELD_SESSIONID = "sessionId.keyword";
    public static final String FIELD_DATAID = "dataId.keyword";
    public static final String FIELD_SIZE = "size.keyword";
    public static final String FIELD_PAGENUM = "pagenum.keyword";
    public static final String FIELD_LOGID = "logId.keyword";
    public static final String FILED_LEVEL = "level.keyword";
    public static final String FIELD_URI = "uri.keyword";
    public static final String FIELD_CODE = "code";

    private static final Logger LOGGER = LoggerFactory.getLogger(EsService.class);

    /**
     * 获取某个行为总的文档数
     *
     * @param action
     *            行为类型
     * @param from
     *            开始日期
     * @param to
     *            结束日期
     * @return
     */
    public long getActionCount(Action action, double from, double to) throws Exception {
        long count = -1;
        SearchResponse response = esBaseQuery.countRange(SERVICE_KEY, action.toString(), from, to);
        int failedShards = response.getFailedShards();
        if (failedShards > 0) {
            throw new EsRespException("failed shards: " + failedShards);
        }
        RestStatus restStatus = response.status();
        if (restStatus.getStatus() != 200) {
            throw new EsRespException("elasticsearch response status err, code: " + restStatus.getStatus());
        }
        Range.Bucket rangeBucket = getFirstBucket(response);
        Cardinality cardinalityAgg = rangeBucket.getAggregations().get("cardinality");
        count = cardinalityAgg.getValue();
        if (count < 0) {
            throw new StatisticCountException("search count shout not be little than zero");
        }

        return count;
    }

    /**
     * 获取某个行为加了限制条件的文档数
     *
     * @param action
     * @param from
     * @param to
     * @param boolQueryBuilder
     *            限制查询条件
     * @return
     */
    public long getActionCount(Action action, double from, double to, BoolQueryBuilder boolQueryBuilder)
            throws Exception {
        long count = -1;
        if (boolQueryBuilder == null) {
            return getActionCount(action, from, to);
        }
        SearchResponse response = esBaseQuery.countRange(SERVICE_KEY, action.toString(), from, to, boolQueryBuilder);
        int failedShards = response.getFailedShards();
        if (failedShards > 0) {
            throw new EsRespException("failed shards: " + failedShards);
        }
        RestStatus restStatus = response.status();
        if (restStatus.getStatus() != 200) {
            throw new EsRespException("elasticsearch response status err, code: " + restStatus.getStatus());
        }
        Range.Bucket rangeBucket = getFirstBucket(response);
        if (null == rangeBucket) {
            throw new StatisticCountException("response is empty");
        }

        if (Action.CLICK.toString().equals(action.toString())) {
            count = rangeBucket.getDocCount();
        }else {
            count = getAggValue(rangeBucket);
        }
        if (count < 0) {
            throw new StatisticCountException("search count shout not be little than zero");
        }

        return count;
    }

    /**
     * 获取聚合结果中文档的数量
     * @param rangeBucket
     * @return
     */
    private long getAggValue(Range.Bucket rangeBucket) {
        Aggregations aggregations = rangeBucket.getAggregations();
        List<Aggregation> list = aggregations.asList();
        Aggregation aggregation = list.get(0);
        String json = JacksonUtils.toJson(aggregation);
        JsonNode jsonNode = JacksonUtils.toJsonNode(json);
        return jsonNode.get("value").asLong();
    }

    /**
     * 获取某个行为加了限制条件的文档数,以天为维度统计
     *
     * @param action
     * @param from
     * @param to
     * @param interval
     *            统计的维度，day|week|month
     * @param boolQueryBuilder
     * @return
     * @throws Exception
     */
    public List<DateCount> getActionHistogram(Action action, double from, double to, String interval,
                                              BoolQueryBuilder boolQueryBuilder) throws Exception {
        List<DateCount> counts = new ArrayList<>();
        String[] cardinalities = new String[]{FIELD_LOGID};
        SearchResponse response = esBaseQuery.countHistogram(SERVICE_KEY, action.toString(), cardinalities, from, to,
                interval, boolQueryBuilder);
        int failedShards = response.getFailedShards();
        if (failedShards > 0) {
            printFailedShard(response);
            throw new EsRespException("failed shards: " + failedShards);
        }
        RestStatus restStatus = response.status();
        if (restStatus.getStatus() != 200) {
            throw new EsRespException("elasticsearch response status err, code: " + restStatus.getStatus());
        }
        Range.Bucket rangeBucket = getFirstBucket(response);
        Histogram histogramAgg = rangeBucket.getAggregations().get("histogram");
        if (histogramAgg == null) {
            throw new IOException("no data in elasticsearch");
        }
        for (Histogram.Bucket bucket : histogramAgg.getBuckets()) {
            String date = Utils.parseDate(bucket.getKeyAsString());
            Cardinality cardinalityAgg = bucket.getAggregations().get("cardinality");
            DateCount count;
            if (cardinalityAgg != null) {
                long docCount = cardinalityAgg.getValue();
                count = new DateCount(date, docCount);
            } else {
                count = new DateCount(date, bucket.getDocCount());
            }
            counts.add(count);
        }
        return counts;
    }

    /**
     * 获取某个行为加了限制条件的文档数,以天为维度统计,对cardinalities做了进一步去重
     *
     * @param action
     * @param cardinalities
     * @param from
     * @param to
     * @param interval
     * @param boolQueryBuilder
     * @return
     * @throws Exception
     */
    public List<DateCount> getActionHistogram(Action action, String[] cardinalities, double from, double to,
            String interval, BoolQueryBuilder boolQueryBuilder) throws Exception {
        List<DateCount> counts = new ArrayList<>();
        SearchResponse response = null;
        try {
            response = esBaseQuery.countHistogram(SERVICE_KEY, action.toString(), cardinalities, from, to, interval,
                    boolQueryBuilder);
        } catch (Exception e) {
            LOGGER.error("getActionHistogram failed, err: ", e);
            return counts;
        }

        if (response == null) {
            throw new Exception("es request failed, response is null");
        }

        int failedShards = response.getFailedShards();
        if (failedShards > 0) {
            printFailedShard(response);
            throw new EsRespException("failed shards: " + failedShards);
        }
        RestStatus restStatus = response.status();
        if (restStatus.getStatus() != 200) {
            throw new EsRespException("elasticsearch response status err, code: " + restStatus.getStatus());
        }
        Range.Bucket rangeBucket = getFirstBucket(response);
        Histogram histogramAgg = rangeBucket.getAggregations().get("histogram");
        if (histogramAgg == null) {
            throw new IOException("no data in elasticsearch");
        }
        for (Histogram.Bucket bucket : histogramAgg.getBuckets()) {
            String date = Utils.parseDate(bucket.getKeyAsString());
            DateCount count;
            Cardinality cardinalityAgg = bucket.getAggregations().get("cardinality");
            if (cardinalityAgg != null) {
                long docCount = cardinalityAgg.getValue();
                count = new DateCount(date, docCount);
            } else {
                count = new DateCount(date, bucket.getDocCount());
            }
            counts.add(count);
        }
        return counts;
    }

    /**
     * 获取某个行为加了多个限制条件的文件数
     *
     * @param action
     * @param from
     * @param to
     * @param queryBuilders
     * @return
     */
    public List<Long> getMultiActionCount(Action action, double from, double to, QueryBuilder... queryBuilders)
            throws Exception {
        List<Long> counts = new ArrayList<>(queryBuilders.length);
        MultiSearchResponse multiSearchResponse = esBaseQuery.multiCountRange(SERVICE_KEY, action.toString(), from, to,
                queryBuilders);

        for (MultiSearchResponse.Item item : multiSearchResponse.getResponses()) {
            SearchResponse response = item.getResponse();
            System.out.println(response == null);
            int failedShards = response.getFailedShards();
            if (failedShards > 0) {
                throw new EsRespException("failed shards: " + failedShards);
            }
            RestStatus restStatus = response.status();
            if (restStatus.getStatus() != 200) {
                throw new EsRespException("elasticsearch response status err, code: " + restStatus.getStatus());
            }
            Range.Bucket rangeBucket = getFirstBucket(response);
            long count = rangeBucket.getDocCount();
            if (count < 0) {
                throw new StatisticCountException("search count shout not be little than zero");
            }
            counts.add(count);
        }

        return counts;
    }

    /**
     * 获取出现次数前几的关键词
     *
     * @param action
     *            行为类型
     * @param filed
     *            关键词对应的字段
     * @param from
     *            开始日期
     * @param to
     *            结束日期
     * @param size
     * @param queryBuilder
     *            附加查询条件
     * @return
     */
    public Range.Bucket getTopWords(Action action, String filed, double from, double to, int size,
            QueryBuilder queryBuilder) throws Exception {
        SearchResponse response = esBaseQuery.aggregationTerm(SERVICE_KEY, action.toString(), filed, FIELD_LOGID, from,
                to, size, queryBuilder);

        return getFirstBucket(response);
    }

    public Range.Bucket getMultiTopWords(Action[] actions, String script, String cardCondition, double from, double to,
            int size) throws Exception {
        String[] arr = new String[actions.length];
        int count = 0;
        for (Action action : actions) {
            arr[count] = action.toString();
            count++;
        }
        SearchResponse response = esBaseQuery.aggregationTerms(SERVICE_KEY, arr, script, SERVICE_KEY, cardCondition,
                from, to, size);
        return getFirstBucket(response);
    }

    /**
     * 获取去重后的数量
     *
     * @param action
     * @param filed
     * @param from
     * @param to
     * @param queryBuilder
     * @return
     * @throws Exception
     */
    public long getDistinctCount(Action action, String filed, double from, double to, QueryBuilder queryBuilder)
            throws Exception {
        return getDistinctCount(action, new String[]{filed}, from, to, queryBuilder);
    }

    public long getDistinctCount(Action action, String[] fields, double from, double to, QueryBuilder queryBuilder)
            throws Exception {
        long count = -1;
        SearchResponse response = esBaseQuery.aggregationCardinality(SERVICE_KEY, action.toString(), fields, from, to,
                queryBuilder);
        Range.Bucket rangeBucket = getFirstBucket(response);
        Aggregation cardinalityAgg = rangeBucket.getAggregations().get("cardinality");
        if (cardinalityAgg == null) {
            throw new IOException("no data in elasticsearch");
        }
        count = ((Cardinality) cardinalityAgg).getValue();
        if (count < 0) {
            throw new StatisticCountException("agg count shout not be little than zero");
        }

        return count;
    }

    public Range.Bucket getFirstBucket(SearchResponse searchResponse) throws Exception {
        if (getRangeBuckets(searchResponse).size() > 0) {
            return getRangeBuckets(searchResponse).get(0);
        }
        return null;
    }

    public List<? extends Range.Bucket> getRangeBuckets(SearchResponse searchResponse) throws Exception {
        Aggregation rangeAgg = searchResponse.getAggregations().get("range");
        List<? extends Range.Bucket> buckets = ((Range) rangeAgg).getBuckets();
        if (buckets.size() < 1) {
            throw new StatisticCountException("search err, buckets is null");
        }
        return buckets;
    }

    public void printFailedShard(SearchResponse response) {
        ShardSearchFailure[] shardFailures = response.getShardFailures();
        List<Throwable> errList = new ArrayList<>(response.getFailedShards());
        for (ShardSearchFailure shardFailure : shardFailures) {
            errList.add(shardFailure.getCause());
        }
        LOGGER.error("failed err list: ", errList);
    }

    public void close() {
        try {
            esBaseQuery.close();
        } catch (IOException e) {
            LOGGER.debug("close es service failed, err: " + e);
        }
    }
}
